package com.alibaba.otter.node.etl.load.loader.mq.rabbit.api;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqConstants;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqUtils;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.ConnectedCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.DefaultConnectedCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.DefaultDisconnectedCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.DisconnectedCallback;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.recovery.AutorecoveringChannel;
import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;

/**
 * Channel管理类，RabbitMQ采用Channel的方式来复用连接，一个Connection对应多个Channel。
 *
 * @author dengchao
 */
@Slf4j
public class ChannelManager {

    @Setter
    private long retryInterval = MqConstants.RECREATE_CHANNEL_INTERVAL_IN_MS;

    @Setter
    @Getter
    private int prefetchCount = MqConstants.DEFAULT_PREFETCH_SIZE;

    @Setter
    private ConnectedCallback connectedCallback = new DefaultConnectedCallback();

    @Setter
    private DisconnectedCallback disconnectedCallback = new DefaultDisconnectedCallback();

    private static LoadingCache<String, ChannelPool> publisherChannelCache = CacheBuilder.newBuilder()
            .build(new CacheLoader<String, ChannelPool>() {
                @Override
                public ChannelPool load(String namespaceName) {
                    return new ChannelPool(namespaceName);
                }
            });

    private static RabbitMqConnectionManager connectionManager = new RabbitMqConnectionManager();

    private LoadingCache<String, AutorecoveringChannel> listenerChannelCache = CacheBuilder.newBuilder()
            .build(new CacheLoader<String, AutorecoveringChannel>() {
                @Override
                public AutorecoveringChannel load(String namespaceName) throws Exception {
                    return (AutorecoveringChannel) connectionManager.getConnection(namespaceName).createChannel();
                }
            });

    public AutorecoveringConnection getConn(String namespaceName) {
        return (AutorecoveringConnection) connectionManager.getConnection(namespaceName);
    }


    /**
     * 获取指定命名空间下的channel, channel维护在channel池中。
     *
     * @param namespaceName 命名空间
     * @return 活跃的channel
     */
    public Channel getPublisherChannel(String namespaceName) {
        try {
            ChannelPool pool = publisherChannelCache.get(namespaceName);
            pool.setConnectedCallback(connectedCallback);
            pool.setDisconnectedCallback(disconnectedCallback);

            Channel channel = pool.getChannel();
            channel.basicQos(prefetchCount);
            return channel;
        } catch (IOException | ExecutionException e) {
            log.error("创建channel失败，" + retryInterval + "毫秒后重试。" + e.getMessage(), e);
            MqUtils.sleep(retryInterval);
            return getPublisherChannel(namespaceName);
        }
    }

    /**
     * 获取指定命名空间下的channel。
     *
     * @param namespaceName 命名空间
     * @return 活跃的channel
     */
    public AutorecoveringChannel getListenerChannel(String namespaceName) {
        try {
            namespaceName = namespaceName == null ? "" : namespaceName;
            AutorecoveringChannel channel = listenerChannelCache.get(namespaceName);
            while (!channel.isOpen()) {
                listenerChannelCache.invalidate(namespaceName);
                channel = listenerChannelCache.get(namespaceName);
                MqUtils.sleep(retryInterval);
            }
            channel.basicQos(prefetchCount);
            connectedCallback.connectedCallback(channel);
            return channel;
        } catch (IOException | ExecutionException e) {
            log.error("创建channel失败，" + retryInterval + "毫秒后重试。" + e.getMessage(), e);
            MqUtils.sleep(retryInterval);
            return getListenerChannel(namespaceName);
        }
    }

    protected Collection<Channel> listChannels() {
        List<Channel> channels = new ArrayList<>();
        for (ChannelPool container : publisherChannelCache.asMap().values()) {
            channels.addAll(container.getAllChannels().values());
        }
        return channels;
    }

    public void releaseChannel(Channel channel, String namespace) {
        if (channel == null) {
            return;
        }
        try {
            publisherChannelCache.get(namespace).release(channel);
        } catch (ExecutionException e) {
            log.warn("释放channel出错，直接关闭channel。", e);
            try {
                channel.close();
            } catch (IOException | TimeoutException e1) {
                log.warn("关闭channel出错。", e);
            }
        }
    }

}
